from xtquant import xtdata
import time
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional


class MACDConfig:
    """MACD性能配置类"""

    # 基础配置
    USE_WARMUP = False  # 是否使用预热数据（同花顺风格默认不使用）
    USE_THS_STYLE = True  # 是否使用同花顺风格计算（默认True）
    PREFER_PREVIOUS_DAY_TICKS = True  # 优先使用前一天的tick数据作为预热数据
    WARMUP_DAYS = 6  # 预热数据天数
    WARMUP_HOURS = 4  # 预热数据小时数

    # 性能优化配置
    TICK_CHECK_INTERVAL = 1  # 每N个tick检查一次MACD
    CACHE_TTL = 30  # MACD结果缓存有效期（秒）
    MAX_CACHE_SIZE = 300  # 最大缓存条目数

    # 计算参数
    FAST_EMA = 8  # 快线EMA周期
    SLOW_EMA = 17  # 慢线EMA周期
    SIGNAL_EMA = 9  # 信号线EMA周期
    CROSS_EPSILON = 1e-5  # 金叉/死叉判定容差，防止0附近抖动。想更稳：把CROSS_EPSILON调大一点（如5e-6或1e-5），把MIN_CROSS_GAP_BARS设为2~3。
    MIN_CROSS_GAP_BARS = 10  # 两次交叉最少间隔bar数，防止连续抖动重复判定

    # 信号检测配置
    SIGNAL_TIME_WINDOW = 300  # 信号检测时间窗口（秒）
    MIN_DATA_POINTS = 50  # 最小数据点要求

    # 输出控制
    LOG_LEVEL = "INFO"  # 日志级别：DEBUG, INFO, WARNING, ERROR
    ENABLE_DETAILED_LOGGING = False  # 是否启用详细日志

    # 资源限制
    MAX_CONCURRENT_CALCULATIONS = 3  # 最大并发计算数
    CALCULATION_TIMEOUT = 30  # 计算超时时间（秒）

    @classmethod
    def get_optimized_params(cls):
        """获取优化后的参数配置"""
        return {
            'use_warmup': cls.USE_WARMUP,
            'use_ths_style': cls.USE_THS_STYLE,
            'prefer_previous_day_ticks': cls.PREFER_PREVIOUS_DAY_TICKS,
            'warmup_days': cls.WARMUP_DAYS,
            'warmup_hours': cls.WARMUP_HOURS,
            'fast_ema': cls.FAST_EMA,
            'slow_ema': cls.SLOW_EMA,
            'signal_ema': cls.SIGNAL_EMA,
            'cache_ttl': cls.CACHE_TTL,
            'max_cache_size': cls.MAX_CACHE_SIZE
        }

    @classmethod
    def should_check_macd(cls, tick_count: int) -> bool:
        """判断是否应该检查MACD"""
        return tick_count % cls.TICK_CHECK_INTERVAL == 0

    @classmethod
    def is_cache_valid(cls, last_check_time: float, current_time: float) -> bool:
        """判断缓存是否有效"""
        return (current_time - last_check_time) < cls.CACHE_TTL

    @classmethod
    def get_signal_window(cls) -> int:
        """获取信号检测时间窗口"""
        return cls.SIGNAL_TIME_WINDOW


# 缓存管理配置
class CacheConfig:
    """缓存管理配置"""

    ENABLE_CACHE = True
    CACHE_CLEANUP_INTERVAL = 300  # 缓存清理间隔（秒）
    CACHE_STATS_LOG_INTERVAL = 600  # 缓存统计日志间隔（秒）

    @classmethod
    def should_use_cache(cls) -> bool:
        """是否应该使用缓存"""
        return cls.ENABLE_CACHE

    @classmethod
    def get_cache_config(cls):
        """获取缓存配置"""
        return {
            'enable_cache': cls.ENABLE_CACHE,
            'cleanup_interval': cls.CACHE_CLEANUP_INTERVAL,
            'stats_log_interval': cls.CACHE_STATS_LOG_INTERVAL
        }


##MACD信号计算器类
class MACDSignalCalculator:

    def __init__(self, xtdata, use_warmup: bool = True):
        """
        初始化MACD信号计算器

        Args:
            use_warmup: 是否使用预热数据，默认True
        """
        self.use_warmup = use_warmup
        self.xtdata = xtdata

    ##获取前一天的tick数据用于预热
    def get_previous_day_ticks_for_warmup(self, stock_code: str) -> Optional[pd.DataFrame]:
        """
        获取前一天的tick数据用于预热，只取下午1点到3点的数据

        Args:
            stock_code: 股票代码

        Returns:
            过滤后的前一天下午tick数据DataFrame
        """
        try:
            # 获取前一个交易日的日期（如果是周一则获取上周五）
            today = datetime.now()
            if today.weekday() == 0:  # 周一
                # 获取上周五的日期
                yesterday = (today - timedelta(days=7)).strftime("%Y%m%d")
            else:
                # 获取前一天的日期
                yesterday = (today - timedelta(days=7)).strftime("%Y%m%d")

            # 下载前一天的tick数据
            self.xtdata.download_history_data(stock_code=stock_code, period="tick",
                                              start_time=yesterday, end_time=yesterday)
            time.sleep(2)

            # 获取本地数据
            data = self.xtdata.get_local_data(stock_list=[stock_code], period='tick',
                                              start_time=yesterday, end_time=yesterday)

            if not data or stock_code not in data:
                return None

            df = data[stock_code]
            if df.empty:
                return None

            # 确定时间列和价格列
            time_col = next((col for col in df.columns if 'time' in col.lower()), None)
            price_col = next((col for col in ['price', 'lastPrice', 'close', 'last', 'current']
                              if col in df.columns), None)

            if not time_col or not price_col:
                return None

            # 转换时间戳
            try:
                if df[time_col].dtype in ['int64', 'float64']:
                    df['time'] = pd.to_datetime(df[time_col], unit='ms') + pd.Timedelta(hours=8)
                else:
                    df['time'] = pd.to_datetime(df[time_col], format='%Y%m%d%H%M%S', errors='coerce')

                df = df[df['time'].notna()].sort_values('time')
                if df.empty:
                    return None
            except Exception:
                return None

            # 过滤无效价格数据
            df_filtered = df[df[price_col] > 0].copy()
            if df_filtered.empty:
                return None

            df_filtered['time_only'] = df_filtered['time'].dt.time

            # 只保留交易时间的数据作为预热数据：9:30-11:30 和 13:00-15:00
            morning_start = pd.Timestamp('09:30:00').time()
            morning_end = pd.Timestamp('11:30:00').time()
            afternoon_start = pd.Timestamp('13:00:00').time()
            afternoon_end = pd.Timestamp('15:00:00').time()

            df_filtered = df_filtered[
                ((df_filtered['time_only'] >= morning_start) & (df_filtered['time_only'] <= morning_end)) |
                ((df_filtered['time_only'] >= afternoon_start) & (df_filtered['time_only'] <= afternoon_end))
                ]

            # 调试信息：显示预热数据的时间范围
            if not df_filtered.empty:
                print(
                    f"预热数据时间范围: {df_filtered['time'].min().strftime('%H:%M')} 到 {df_filtered['time'].max().strftime('%H:%M')}")
                print(f"预热数据点数量: {len(df_filtered)}")
                print("预热数据包含: 9:30-11:30 和 13:00-15:00 交易时间")

            if df_filtered.empty:
                return None

            # 只保留必要的列
            result_df = df_filtered[['time', price_col]].copy()
            result_df.rename(columns={price_col: 'close'}, inplace=True)

            return result_df

        except Exception as e:
            print(f"获取前一天tick数据失败: {e}")
            return None

    ##获取前N天的日线数据用于预热（保留原有方法作为备选）
    def get_historical_data_for_warmup(self, stock_code: str, days: int = 5) -> Optional[pd.DataFrame]:
        try:
            # 获取前一个交易日的日期（如果是周一则获取上周五）
            today = datetime.now()
            if today.weekday() == 0:  # 周一
                # 获取上周五的日期
                end_date = (today - timedelta(days=3)).strftime("%Y%m%d")
            else:
                # 获取前一天的日期
                end_date = (today - timedelta(days=1)).strftime("%Y%m%d")

            start_date = (today - timedelta(days=days)).strftime("%Y%m%d")

            self.xtdata.download_history_data(stock_code=stock_code, period="1d", start_time=start_date,
                                              end_time=end_date)
            time.sleep(1)

            data = self.xtdata.get_local_data(stock_list=[stock_code], period='1d', start_time=start_date,
                                              end_time=end_date)
            if data and stock_code in data:
                return data[stock_code]
            return None
        except Exception:
            return None

    ##生成预热数据，模拟开盘前的价格波动（保留原有方法作为备选）
    def generate_warmup_data(self, latest_close: float, warmup_hours: int = 2) -> pd.DataFrame:
        # 从8:00开始，每5分钟一个数据点
        warmup_points = warmup_hours * 12  # 2小时 * 12个5分钟点

        # 基于最新收盘价生成预热数据
        base_price = latest_close

        # 生成时间序列（从8:00开始）
        start_time = datetime.now().replace(hour=8, minute=0, second=0, microsecond=0)
        warmup_times = [start_time + timedelta(minutes=5 * i) for i in range(warmup_points)]

        # 生成价格数据（模拟小幅波动）
        np.random.seed(42)  # 固定随机种子，确保可重现
        price_changes = np.random.normal(0, base_price * 0.001, warmup_points)  # 0.1%的标准差
        warmup_prices = [base_price + change for change in price_changes]

        # 创建预热数据DataFrame
        warmup_df = pd.DataFrame({
            'time': warmup_times,
            'price': warmup_prices,
            'open': warmup_prices,
            'high': [p * (1 + abs(np.random.normal(0, 0.0005))) for p in warmup_prices],
            'low': [p * (1 - abs(np.random.normal(0, 0.0005))) for p in warmup_prices],
            'close': warmup_prices,
            'volume': [int(np.random.uniform(100, 1000)) for _ in range(warmup_points)]
        })

        return warmup_df

    ##同花顺风格的EMA计算，使用SMA初始化
    def ths_style_ema(self, series: pd.Series, period: int) -> pd.Series:
        """
        同花顺风格的EMA计算：
        - 当数据点数量 <= period时，使用SMA（简单移动平均）
        - 当数据点数量 > period时，使用标准EMA计算

        Args:
            series: 价格序列
            period: EMA周期

        Returns:
            EMA序列
        """
        if not isinstance(series, pd.Series):
            series = pd.Series(series)

        if series.empty:
            return pd.Series([], dtype=float)

        # 重置索引以便按位置访问
        series = series.reset_index(drop=True)
        ema_values = []

        for i in range(len(series)):
            if i < period - 1:
                # 数据不足period个，使用SMA
                sma_value = series.iloc[:i + 1].mean()
                ema_values.append(sma_value)
            elif i == period - 1:
                # 第period个点，用SMA初始化EMA
                sma_init = series.iloc[:period].mean()
                ema_values.append(sma_init)
            else:
                # 第period+1个点开始，使用标准EMA计算
                alpha = 2 / (period + 1)
                new_ema = alpha * series.iloc[i] + (1 - alpha) * ema_values[i - 1]
                ema_values.append(new_ema)

        return pd.Series(ema_values, index=series.index)

    ##支持预热数据的EMA计算，实现热启动（保留原方法作为备选）
    def enhanced_ema_with_warmup(self, series: pd.Series, period: int,
                                 warmup_data: Optional[pd.DataFrame] = None) -> pd.Series:
        if not isinstance(series, pd.Series):
            series = pd.Series(series)

        if warmup_data is not None and not warmup_data.empty:
            # 使用预热数据初始化EMA
            warmup_series = warmup_data['close'] if 'close' in warmup_data.columns else warmup_data['price']
            if len(warmup_series) >= period:
                # 从预热数据计算初始EMA值
                initial_ema = warmup_series.tail(period).mean()
                # 从预热数据的最后一个值开始计算
                last_warmup_price = warmup_series.iloc[-1]
            else:
                # 预热数据不足，使用简单平均
                initial_ema = warmup_series.mean() if not warmup_series.empty else series.iloc[0]
                last_warmup_price = initial_ema
        else:
            # 无预热数据，使用传统方法
            initial_ema = series.iloc[0]
            last_warmup_price = initial_ema

        # 计算EMA
        alpha = 2 / (period + 1)
        ema_values = [initial_ema]

        for price in series:
            new_ema = alpha * price + (1 - alpha) * ema_values[-1]
            ema_values.append(new_ema)

        # 返回EMA值（去掉初始值）
        return pd.Series(ema_values[1:], index=series.index)

    ##同花顺风格的MACD计算，不使用预热数据
    def ths_style_calculate_macd(self, prices: pd.Series, fast: int = 8, slow: int = 17, signal: int = 9) -> Tuple[
        pd.Series, pd.Series, pd.Series]:
        """
        同花顺风格的MACD计算：
        - 不使用预热数据，从当天开盘开始计算
        - EMA使用SMA初始化，符合主流软件做法

        Args:
            prices: 价格序列
            fast: 快线EMA周期
            slow: 慢线EMA周期  
            signal: 信号线EMA周期

        Returns:
            (DIF, DEA, MACD柱状图)
        """
        if not isinstance(prices, pd.Series):
            prices = pd.Series(prices)

        if prices.empty:
            return pd.Series([], dtype=float), pd.Series([], dtype=float), pd.Series([], dtype=float)

        # 使用同花顺风格的EMA计算
        ema_fast = self.ths_style_ema(prices, fast)
        ema_slow = self.ths_style_ema(prices, slow)

        # 计算DIF
        dif = ema_fast - ema_slow

        # 计算DEA（信号线），也使用同花顺风格
        dea = self.ths_style_ema(dif, signal)

        # 计算MACD柱状图
        macd_histogram = 2 * (dif - dea)

        return dif, dea, macd_histogram

    ##支持预热数据的MACD计算，实现热启动"
    def enhanced_calculate_macd(self, prices: pd.Series, fast: int = 8, slow: int = 17, signal: int = 9,
                                warmup_data: Optional[pd.DataFrame] = None) -> Tuple[pd.Series, pd.Series, pd.Series]:
        if not isinstance(prices, pd.Series):
            prices = pd.Series(prices)

        # 确保为float并使用顺序索引
        series = prices.astype(float).reset_index(drop=True)

        # 使用增强版EMA计算
        ema_short = self.enhanced_ema_with_warmup(series, fast, warmup_data)
        ema_slow = self.enhanced_ema_with_warmup(series, slow, warmup_data)

        # 计算DIFF
        diff = ema_short - ema_slow

        # 使用DIFF计算DEA（信号线）
        if warmup_data is not None and not warmup_data.empty:
            # 如果有预热数据，使用预热数据计算预热DIFF来初始化DEA
            # 确定预热数据的价格列
            warmup_price_col = next((col for col in ['close', 'price'] if col in warmup_data.columns), None)
            if warmup_price_col and len(warmup_data[warmup_price_col]) >= max(fast, slow):
                # 计算预热数据的EMA
                warmup_ema_short = self.enhanced_ema_with_warmup(warmup_data[warmup_price_col], fast)
                warmup_ema_slow = self.enhanced_ema_with_warmup(warmup_data[warmup_price_col], slow)
                # 计算预热数据的DIFF
                warmup_diff_series = warmup_ema_short - warmup_ema_slow
                # 使用预热数据DIFF的最后一个值来初始化DEA
                warmup_diff_last = warmup_diff_series.iloc[-1] if not warmup_diff_series.empty else 0
            else:
                warmup_diff_last = 0

            # 使用预热DIFF的最后一个值来初始化DEA计算
            dea = self.enhanced_ema_with_warmup(diff, signal, pd.DataFrame({'close': [warmup_diff_last]}))
        else:
            dea = self.enhanced_ema_with_warmup(diff, signal)

        # 计算MACD柱状图
        macd_histogram = 2 * (diff - dea)

        return diff, dea, macd_histogram

    ##将tick数据重新采样为分钟数据
    def resample_ticks_to_minutes(self, df: pd.DataFrame, price_col: str = 'price') -> pd.DataFrame:
        if 'time' not in df.columns:
            return pd.DataFrame()

        df_resampled = df.set_index('time')
        df_minutes = df_resampled[price_col].resample('1min').agg({
            'open': 'first',
            'high': 'max',
            'low': 'min',
            'close': 'last',
            'volume': 'sum' if 'volume' in df_resampled.columns else lambda x: len(x)
        }).dropna()
        return df_minutes.reset_index()

    ##下载并获取tick数据
    def get_segment_ticks(self, stock_code: str) -> Optional[Dict[str, pd.DataFrame]]:
        today = datetime.now().strftime("%Y%m%d")
        # today = "20250811"
        self.xtdata.download_history_data(stock_code=stock_code, period="tick", start_time=today, end_time=today)
        time.sleep(2)

        try:
            data = self.xtdata.get_local_data(stock_list=[stock_code], period='tick', start_time=today, end_time=today)
            return data if data and stock_code in data else None
        except Exception:
            return None

    ##传统MACD计算函数，支持预热数据
    def calculate_macd(self, prices: pd.Series, fast: int = 8, slow: int = 17, signal: int = 9,
                       warmup_data: Optional[pd.DataFrame] = None) -> Tuple[pd.Series, pd.Series, pd.Series]:
        if warmup_data is not None and not warmup_data.empty:
            # 使用增强版MACD计算
            return self.enhanced_calculate_macd(prices, fast, slow, signal, warmup_data)

        if not isinstance(prices, pd.Series):
            prices = pd.Series(prices)
        # 确保为float并使用顺序索引，便于逐步递推
        series = prices.astype(float).reset_index(drop=True)

        def ths_ema(series_in: pd.Series, period: int) -> pd.Series:
            if len(series_in) < period:
                return pd.Series([np.nan] * len(series_in), index=series_in.index)
            ema = series_in.astype(float).copy()
            # 前period-1为NaN，第period个点为SMA初始化
            ema.iloc[: period - 1] = np.nan
            sma_init = series_in.iloc[:period].mean()
            ema.iloc[period - 1] = sma_init
            alpha = 2 / (period + 1)
            for i in range(period, len(series_in)):
                ema.iloc[i] = alpha * series_in.iloc[i] + (1 - alpha) * ema.iloc[i - 1]
            return ema

        ema_short = ths_ema(series, fast)
        ema_slow = ths_ema(series, slow)
        diff = ema_short - ema_slow
        valid_diff = diff.dropna()
        if len(valid_diff) < signal:
            dea_full = pd.Series([np.nan] * len(series))
            macd_full = pd.Series([np.nan] * len(series))
            return diff, dea_full, macd_full
        dea_valid = ths_ema(valid_diff, signal)
        dea_full = pd.Series([np.nan] * len(series))
        dea_full.iloc[dea_valid.index] = dea_valid.values
        macd_valid = 2 * (valid_diff.loc[dea_valid.index] - dea_valid)
        macd_full = pd.Series([np.nan] * len(series))
        macd_full.iloc[macd_valid.index] = macd_valid.values
        return diff, dea_full, macd_full

    ##检测MACD金叉死叉信号并返回结果
    def detect_macd_signals(self, stock_code: str, fast: int = 8, slow: int = 17, signal: int = 9,
                            use_warmup: Optional[bool] = None, use_ths_style: bool = True) -> Dict[
        str, List[Dict[str, any]]]:
        """
        检测MACD金叉死叉信号并返回结果

        Args:
            stock_code: 股票代码
            fast: 快线EMA周期
            slow: 慢线EMA周期
            signal: 信号线EMA周期
            use_warmup: 是否使用预热数据（当use_ths_style=True时忽略）
            use_ths_style: 是否使用同花顺风格计算（默认True）

        Returns:
            包含金叉和死叉信号的字典，格式如下：
            {
                'golden_cross': [{'time': '2024-01-01 09:30:00', 'price': 10.5}, ...],
                'death_cross': [{'time': '2024-01-01 10:15:00', 'price': 10.3}, ...],
                'latest_macd': {'diff': 0.001, 'dea': -0.002, 'macd': 0.006, 'trend': 'bullish'},
                'data_info': {'total_minutes': 240, 'time_range': '09:30-15:00'}
            }
        """
        # 获取tick数据
        ticks = self.get_segment_ticks(stock_code)
        if not ticks or stock_code not in ticks:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        df = ticks[stock_code]
        if df.empty:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        # 预热数据处理 - 根据配置选择预热数据源
        warmup_df = None
        use_warmup_effective = self.use_warmup if use_warmup is None else use_warmup

        # 如果使用同花顺风格，则忽略预热数据
        if use_ths_style:
            warmup_df = None
            print("使用同花顺风格MACD计算（不使用预热数据）")
        elif use_warmup_effective:
            try:
                # 使用前一天的tick数据
                warmup_df = self.get_previous_day_ticks_for_warmup(stock_code)
                print("使用预热数据模式")
            except Exception as e:
                print(f"预热数据获取失败，使用无预热模式: {e}")
                warmup_df = None

        # 确定价格列
        price_col = next((col for col in ['price', 'lastPrice', 'close', 'last', 'current'] if col in df.columns), None)
        if not price_col:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        # 转换时间戳
        time_col = next((col for col in df.columns if 'time' in col.lower()), None)
        if not time_col:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        try:
            if df[time_col].dtype in ['int64', 'float64']:
                df['time'] = pd.to_datetime(df[time_col], unit='ms') + pd.Timedelta(hours=8)
            else:
                df['time'] = pd.to_datetime(df[time_col], format='%Y%m%d%H%M%S', errors='coerce')

            df = df[df['time'].notna()].sort_values('time')
            if df.empty:
                return {
                    'golden_cross': [],
                    'death_cross': [],
                    'latest_macd': {},
                    'data_info': {}
                }
        except Exception:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        # 过滤无效数据、集合竞价和午休时间
        df_filtered = df[df[price_col] > 0].copy()

        # 去除集合竞价时间 (9:15-9:25)
        auction_start = pd.Timestamp('09:15:00').time()
        auction_end = pd.Timestamp('09:29:59').time()

        # 去除午休时间 (11:30-13:00)
        lunch_start = pd.Timestamp('11:30:00').time()
        lunch_end = pd.Timestamp('13:00:00').time()

        df_filtered['time_only'] = df_filtered['time'].dt.time

        # 过滤掉集合竞价和午休时间
        df_filtered = df_filtered[
            ~((df_filtered['time_only'] >= auction_start) & (df_filtered['time_only'] <= auction_end)) &
            ~((df_filtered['time_only'] >= lunch_start) & (df_filtered['time_only'] <= lunch_end))
            ]

        if df_filtered.empty:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        # 重新采样为分钟数据
        df_minutes = self.resample_ticks_to_minutes(df_filtered, price_col)
        if df_minutes.empty:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        # 准备预热数据用于MACD计算
        warmup_minutes = None
        if warmup_df is not None:
            warmup_minutes = self.resample_ticks_to_minutes(warmup_df, 'close')

        # 根据参数选择MACD计算方法
        if use_ths_style:
            # 使用同花顺风格计算
            macd_line, signal_line, histogram = self.ths_style_calculate_macd(
                df_minutes['close'], fast=fast, slow=slow, signal=signal
            )
        else:
            # 使用传统方法（支持预热数据）
            macd_line, signal_line, histogram = self.calculate_macd(
                df_minutes['close'], fast=fast, slow=slow, signal=signal, warmup_data=warmup_minutes
            )

        # 检测金叉死叉（鲁棒版，加入容差与斜率确认，避免0附近抖动导致反向提示）
        delta = macd_line - signal_line  # DIF - DEA
        eps = getattr(MACDConfig, 'CROSS_EPSILON', 1e-6)
        min_gap = max(0, int(getattr(MACDConfig, 'MIN_CROSS_GAP_BARS', 1)))

        prev_delta = delta.shift(1)
        delta_change = delta - prev_delta

        # 容差符号：> eps 为正，< -eps 为负，其余视为 0（交汇/近似交汇）
        sign_prev = (prev_delta > eps).astype(int) - (prev_delta < -eps).astype(int)
        sign_now = (delta > eps).astype(int) - (delta < -eps).astype(int)

        # 在存在0的边界时，用斜率来确认方向（只在一侧为0时触发）
        cross_up_mask = (
                ((sign_prev < 0) & (sign_now > 0)) |
                ((sign_prev < 0) & (sign_now == 0) & (delta_change > 0)) |
                ((sign_prev == 0) & (sign_now > 0) & (delta_change > 0))
        )

        cross_down_mask = (
                ((sign_prev > 0) & (sign_now < 0)) |
                ((sign_prev > 0) & (sign_now == 0) & (delta_change < 0)) |
                ((sign_prev == 0) & (sign_now < 0) & (delta_change < 0))
        )

        # 去抖动：限制交叉最小间隔（按bar计）
        def thin_cross(mask_series, min_gap_bars):
            idx = mask_series[mask_series.fillna(False)].index.to_list()
            if not idx:
                return mask_series
            kept = []
            last_kept = None
            for i in idx:
                if last_kept is None or (i - last_kept) >= min_gap_bars:
                    kept.append(i)
                    last_kept = i
            new_mask = mask_series.copy()
            new_mask[:] = False
            new_mask.loc[kept] = True
            return new_mask

        if min_gap > 1:
            cross_up_mask = thin_cross(cross_up_mask, min_gap)
            cross_down_mask = thin_cross(cross_down_mask, min_gap)

        # 构建金叉信号数组
        golden_cross = []
        cross_up_times = df_minutes.loc[cross_up_mask.fillna(False), 'time']
        for t in cross_up_times:
            if pd.isna(t):
                continue
            time_str = pd.to_datetime(t).strftime('%Y-%m-%d %H:%M:%S')
            price_at_time = df_minutes.loc[df_minutes['time'] == t, 'close'].iloc[0] if not df_minutes[
                df_minutes['time'] == t].empty else 0
            golden_cross.append({
                'time': time_str,
                'price': float(price_at_time)
            })

        # 构建死叉信号数组
        death_cross = []
        cross_down_times = df_minutes.loc[cross_down_mask.fillna(False), 'time']
        for t in cross_down_times:
            if pd.isna(t):
                continue
            time_str = pd.to_datetime(t).strftime('%Y-%m-%d %H:%M:%S')
            price_at_time = df_minutes.loc[df_minutes['time'] == t, 'close'].iloc[0] if not df_minutes[
                df_minutes['time'] == t].empty else 0
            death_cross.append({
                'time': time_str,
                'price': float(price_at_time)
            })

        # 构建最新MACD信息（只在有有效数值时返回，避免 NaN 输出）
        latest_macd = {}
        try:
            if not macd_line.empty and not signal_line.empty:
                valid_mask = macd_line.notna() & signal_line.notna()
                if valid_mask.any():
                    last_idx = macd_line[valid_mask].index[-1]
                    latest_diff = macd_line.loc[last_idx]
                    latest_dea = signal_line.loc[last_idx]
                    if not histogram.empty and last_idx in histogram.index and not pd.isna(histogram.loc[last_idx]):
                        latest_histogram = histogram.loc[last_idx]
                    else:
                        latest_histogram = 2 * (latest_diff - latest_dea)

                    latest_macd = {
                        'diff': float(latest_diff),
                        'dea': float(latest_dea),
                        'macd': float(latest_histogram),
                        'trend': 'bullish' if latest_diff > latest_dea else 'bearish'
                    }
        except Exception:
            latest_macd = {}

        # 构建数据信息
        data_info = {
            'total_minutes': len(df_minutes),
            'time_range': f"{df_minutes['time'].min().strftime('%H:%M')}-{df_minutes['time'].max().strftime('%H:%M')}"
        }

        return {
            'golden_cross': golden_cross,
            'death_cross': death_cross,
            'latest_macd': latest_macd,
            'data_info': data_info
        }


# 使用示例
if __name__ == "__main__":
    # 创建数据提供者

    # 创建MACD信号计算器实例
    calculator = MACDSignalCalculator(xtdata, use_warmup=False)  # 默认不使用预热数据

    # 测试预热数据功能
    stock_code = "159529.SZ"
    print(f"测试股票代码: {stock_code}")
    print(f"MACD配置: {MACDConfig.get_optimized_params()}")

    # 测试同花顺风格计算
    print("\n=== 测试同花顺风格MACD计算 ===")
    signals_ths = calculator.detect_macd_signals(stock_code, use_ths_style=True)

    print(f"同花顺风格 - 股票代码: {stock_code}")
    print(f"金叉信号数量: {len(signals_ths['golden_cross'])}")
    print(f"死叉信号数量: {len(signals_ths['death_cross'])}")

    if signals_ths['golden_cross']:
        print("\n金叉信号:")
        for signal in signals_ths['golden_cross']:
            print(f"  时间: {signal['time']}, 价格: {signal['price']}")

    if signals_ths['death_cross']:
        print("\n死叉信号:")
        for signal in signals_ths['death_cross']:
            print(f"  时间: {signal['time']}, 价格: {signal['price']}")

    if signals_ths['latest_macd']:
        print(f"\n最新MACD值:")
        print(f"  DIFF: {signals_ths['latest_macd']['diff']:.6f}")
        print(f"  DEA:  {signals_ths['latest_macd']['dea']:.6f}")
        print(f"  MACD: {signals_ths['latest_macd']['macd']:.6f}")
        print(f"  趋势: {'多头' if signals_ths['latest_macd']['trend'] == 'bullish' else '空头'}")

    print(f"\n数据信息: {signals_ths['data_info']}")

    # 对比测试：传统预热数据方式
    print("\n=== 对比测试：传统预热数据方式 ===")
    signals_traditional = calculator.detect_macd_signals(stock_code, use_ths_style=False, use_warmup=True)

    print(f"传统方式 - 股票代码: {stock_code}")
    print(f"金叉信号数量: {len(signals_traditional['golden_cross'])}")
    print(f"死叉信号数量: {len(signals_traditional['death_cross'])}")

    if signals_traditional['latest_macd']:
        print(f"\n最新MACD值对比:")
        print(f"  同花顺风格 - DIFF: {signals_ths['latest_macd'].get('diff', 0):.6f}")
        print(f"  传统方式   - DIFF: {signals_traditional['latest_macd'].get('diff', 0):.6f}")
        print(f"  同花顺风格 - DEA:  {signals_ths['latest_macd'].get('dea', 0):.6f}")
        print(f"  传统方式   - DEA:  {signals_traditional['latest_macd'].get('dea', 0):.6f}")